package com.atguigu.flink05;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * @author Felix
 * @date 2024/2/23
 * 该案例演示了自定义序列化器将数据写到kafka主题
 * 需求：从指定网络端口读取水位监控数据，将不同传感器采集的数据发送到kafka的不同的主题中
 */
public class Flink01_CustomSchema {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //TODO 2.从指定的网络端口读取数据
        SingleOutputStreamOperator<WaterSensor> wsDS = env
                .socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction());

        //TODO 3.创建kafkaSink对象
        KafkaSink<WaterSensor> sink = KafkaSink.<WaterSensor>builder()
                .setBootstrapServers("hadoop102:9092")
                .setRecordSerializer(
                        new KafkaRecordSerializationSchema<WaterSensor>() {
                            @Nullable
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(WaterSensor ws, KafkaSinkContext context, Long timestamp) {
                                return new ProducerRecord<byte[], byte[]>(ws.id, ws.toString().getBytes());
                            }
                        }
                )

                .build();
        //TODO 4.将流中数据写到kafka中
        wsDS.sinkTo(sink);
        //TODO 5.提交作业

        env.execute();
    }
}
